package com.example.springboot.config.kafka;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.time.LocalDateTime;
import java.util.Map;

/**
 * @Author:ChenZhangKun
 * @Date: 2021/5/25 17:43
 */
public class TimerInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        // 拿到value
        String value = producerRecord.value();
        // 加时间戳
        value = LocalDateTime.now() + value;
        // 返回新的生产记录关系
        ProducerRecord<String, String> producerRecords = new ProducerRecord<String, String>(producerRecord.topic(), producerRecord.key()
                , value);
        return producerRecords;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
